/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.server;

import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;

import org.slf4j.Logger;
import org.sss.common.CommonStatus;
import org.sss.common.Inlet;
import org.sss.mapreduce.Configuration;
import org.sss.mapreduce.Decoder;
import org.sss.mapreduce.Reducer;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.StorageNode;
import org.sss.mapreduce.datatype.Packable;
import org.sss.util.CallbackInterface;

public abstract class ReduceTaskBase<T> extends ProcTask<Reducer, Reducer.Context, List<T>> {
  @SuppressWarnings("unused")
  private static Logger logger = SssServer.getLogger(ReduceTaskBase.class);

  public ReduceTaskBase(TaskSet taskSet,
                        UUID jobID,
                        SssServerResource resource,
                        Configuration conf,
                        StorageNode storageNode,
                        Inlet inlet,
                        Class<? extends Reducer> reducerClass,
                        List<org.sss.common.Outlet> outletList,
                        CallbackInterface<CommonStatus> doneCallback) throws SssException {
    super(taskSet, jobID, resource, conf, storageNode, 
          inlet, reducerClass, outletList, doneCallback,
          resource.getServerConfiguration().reduce_output_cache_nbytes,
          resource.getServerConfiguration().reduce_output_queue_multiplicity,
          resource.getServerConfiguration().reduce_output_queue_limit_nbytes);
  }

  @Override
  protected void configure() throws Exception {
    getProc().configure(createContext());
  }

  @Override
  protected Reducer.Context createContextImpl(ContextImpl impl) {
    return new Context(impl);
  }

  @Override
  protected void submitReadTaskSlices() throws SssException {
    for (int i = 0; i < getInputGID().getNSlot(); i++) {
      submitReadTaskSlice(i);
    }
  }

  protected abstract void submitReadTaskSlice(int subhash) throws SssException;

  private class IterableValues implements Iterable<Packable> {
    private final Decoder valDecoder;
    private final Iterator<byte[]> bytesIt;
    public IterableValues(Decoder decoder, Iterator<byte[]> bytesIt) {
      this.valDecoder = decoder;
      this.bytesIt = bytesIt;
    }
    @Override
    public Iterator<Packable> iterator() {
      return new Iterator<Packable>() {
        @Override
            public boolean hasNext() {
          return bytesIt.hasNext();
        }
        @Override
        public Packable next() {
          try {
            checkCanceled();
            return valDecoder.decode(bytesIt.next());
          }
          catch (SssException e) {
            throw new RuntimeException(e);
          }
          catch (IOException e) {
            throw new RuntimeException(e);
          }
        }
        @Override
        public void remove() {
          throw new UnsupportedOperationException();
        }
      };
    }
  }

  protected IterableValues decodingIterator(Decoder dec, Iterator<byte[]> byteIt) {
    if (byteIt == null) {
      return null;
    }
    return new IterableValues(dec, byteIt);
  }

  private class Context extends ContextBase implements Reducer.Context {
    public Context(ContextImpl impl) {
      super(impl);
    }
  }
}
